package com.zhang.storm.uv;

import com.zhang.storm.pv.PvBolt;
import com.zhang.storm.pv.PvSpout;
import com.zhang.storm.pv.PvSumBolt;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;

public class UvMain {
    public static void main(String[] args) {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("UvSpout", new UvSpout(), 1);
        builder.setBolt("UvBolt", new UvSplitBolt(), 4).shuffleGrouping("UvSpout");
        builder.setBolt("UvSumBolt", new UvSumBolt(), 1).shuffleGrouping("UvBolt");

        Config config = new Config();
        config.setNumWorkers(2);

        if (args.length > 0) {
            try {
                StormSubmitter.submitTopology("UV", config, builder.createTopology());
            } catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
                e.printStackTrace();
            }
        } else {
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology("UV", config, builder.createTopology());
        }
    }
}
